package com.asiainfo.indb.control.db;

import com.asiainfo.indb.control.LogService;
import com.asiainfo.indb.model.InDbGather;
import org.apache.commons.lang3.StringUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * oracle批量入库实现
 *
 * @author zym
 * @date 2019-10-21 15:03
 */
public class OracleBatchFileLineInDb extends FileLineInDb {

    public static final int MAX_BATCH = 10000;
    public static final int DEF_PER_BATCH = 1000;
    private int perBatch;

    /**
     * 数据队列
     */
    private Queue<String[]> dataQueue;

    /**
     * 入表sql
     */
    private String insertSQL;

    /**
     * @param gather 数据上下文
     * @param log    日志
     */
    public OracleBatchFileLineInDb(InDbGather gather, LogService log) {
        super(gather, log);

        // 批次数
        String perBatchStr = gather.get("BATCH_NUMBER");
        if (StringUtils.isNumeric(perBatchStr)) {
            this.perBatch = Integer.parseInt(perBatchStr);
        }

        if (this.perBatch <= 0 || this.perBatch > MAX_BATCH) {
            this.perBatch = DEF_PER_BATCH;
        }

        // 数据队列
        this.dataQueue = new ArrayBlockingQueue<>(perBatch);
    }

    @Override
    public void endInDb() {
        batchInsert(null);

        // 关闭连接
        Connection connection = getConnection();
        if (connection != null) {
            try {
                connection.close();
            } catch (SQLException e) {
                debug("关闭连接异常：" + e.getMessage());
            }
        }
    }

    @Override
    protected void insertDb(String insertSQL, String[] values) {
        this.insertSQL = insertSQL;

        // 放入队列
        if (!dataQueue.offer(values)) {
            batchInsert(values);
        }
    }

    /**
     * 批量入表
     *
     * @param values 值
     */
    private void batchInsert(String[] values) {
        Connection connection = getConnection();
        if (connection != null) {
            try (PreparedStatement preparedStatement = connection.prepareStatement(this.insertSQL)) {
                int len, count = 0;
                String[] oneValue;
                while ((oneValue = dataQueue.poll()) != null) {
                    len = oneValue.length;
                    for (int i = 1; i <= len; i++) {
                        preparedStatement.setObject(i, oneValue[i - 1]);
                    }

                    // 添加批量
                    count++;
                    preparedStatement.addBatch();
                }

                // 提交
                preparedStatement.executeBatch();
                preparedStatement.clearBatch();
                connection.commit();

                debug("提交批次：" + count);
            } catch (SQLException e) {
                debug("入库异常：" + e.getMessage());
            } finally {
                dataQueue.offer(values);
            }
        }
    }
}
